#ifndef __M_MESSAGE_H__
#define __M_MESSAGE_H__
#include "../MQcommon/helper.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include "../MQcommon/logger.hpp"

#include <google/protobuf/map.h>
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <vector>
#include <list>


namespace bitmq{
    #define MAIN_FILE ".mqd"
    #define TMP_FILE  ".mqd.tmp"
    using MessagePtr=std::shared_ptr<bitmq::Message>;
    class MessageMapper{
        public:
        MessageMapper(std::string filedir,const std::string& qname):_queue_name(qname){
            if(filedir.back() != '/'){
                filedir+='/';
            }
            std::string path = FileHelper::parentDirectory(filedir);
            bool ret = FileHelper::createDirectory(path);
            _date_file = filedir+_queue_name+MAIN_FILE;
            _tmp_date_file = filedir+_queue_name+TMP_FILE;
            if(FileHelper::exists(_date_file) == false){
                createMsgFile();
            }
            
        }
        bool createMsgFile(){
            bool ret=FileHelper::createFile(_date_file);
            if(ret == false){
               DLOG("创建文件[%s]失败",_date_file.c_str());
            }
            return ret;
        }
        void removeMsgFile(){
            if(FileHelper::exists(_date_file) == true){
                bool ret = FileHelper::removeFile(_date_file);
                if(ret == true){
                    DLOG("删除文件[%s]成功",_date_file.c_str());
                }
            }
           
            if(FileHelper::exists(_tmp_date_file) == true){
                bool ret = FileHelper::removeFile(_tmp_date_file);
                if(ret == true){
                    DLOG("删除文件[%s]成功",_tmp_date_file.c_str());
                }
            }
        }
        bool insert(MessagePtr &msg){
            return insert(_date_file,msg);
        }
        bool remove(MessagePtr &msg){
            msg->mutable_payload()->set_valid("0");
            std::string body=msg->payload().SerializeAsString();
            int lenth=msg->length();
            size_t offset=msg->offset();
            std::cout<<"当前长度是:"<<lenth<<std::endl;
            if(body.size() != lenth){
                DLOG("数据长度前后不一致，无法删除！");
                return false;
            }
            FileHelper _date_file_helper(_date_file);
            bool ret=_date_file_helper.write(body.c_str(),offset,lenth);
            if(ret == false ){
                DLOG("覆盖原始数据失败！");
                return false;
            }
            return true;
        }
        std::list<MessagePtr> gc(){
             std::list<MessagePtr> result;
             bool ret=load(result);
             if(ret == false){
                DLOG("加载队列[%s]有效数据失败",_queue_name.c_str());
                return result;
             }
            //  DLOG("垃圾回收,得到有效消息数量:%d",result.size());
             if(FileHelper::exists(_tmp_date_file) == false){
                FileHelper::createFile(_tmp_date_file);
             }
             //将有效数据加载到临时文件当中

             for(auto& x:result){
                DLOG("当前有效数据为 %s",x->payload().body().c_str());
                bool ret=insert(_tmp_date_file,x);
                if(ret == false){
                    DLOG("加载有效数据到临时文件失败");
                    return result;
                }
             }
            //  DLOG("当前预加载的文件[%s]大小为 %d",_tmp_date_file.c_str(),FileHelper::size(_tmp_date_file));
             ret = FileHelper::removeFile(_date_file);
             if(ret == false){
                DLOG("删除源文件失败");
             }
             ret = FileHelper(_tmp_date_file).rename(_date_file);
            //  if(ret == true){
            //     DLOG("重命名临时文件成功");
            //  }
             size_t tmp=FileHelper::size(_date_file);
             DLOG("当前预加载的文件[%s]大小为 %d",_date_file.c_str(),tmp);
             return result;
        }
        
        private:
        //加载队列数据文件中的消息
        bool load(std::list<MessagePtr> &result){
            //队列数据文件中的消息数据格式 4字节|消息主体
            FileHelper _date_file_helper(_date_file);
            //将队列中的数据文件全部读取出来，存入临时文件
            size_t offset = 0;//从头开始读
            size_t total_lenth = _date_file_helper.size();//队列数据文件的总长度，读到这里就算全部读完
            size_t lenth = 0;
            // DLOG("当前预加载的文件[%s]大小为 %d",_date_file.c_str(),total_lenth);
            //将消息数据加载出来
            while(offset < total_lenth){
                MessagePtr tmp_ptr = std::make_shared<Message>();
                bool ret = _date_file_helper.read((char*)&lenth,offset,sizeof(size_t));
                if( ret == false){
                    DLOG("加载消息数据长度失败");
                    return false;
                }
                offset+=sizeof(size_t);
                std::string body(lenth, '\0');
                ret = _date_file_helper.read(&(body[0]),offset,lenth);
                if(ret == false){
                    DLOG("加载消息数据失败");
                    return false;
                }
                tmp_ptr->mutable_payload()->ParseFromString(body);
                tmp_ptr->set_length(lenth);
                tmp_ptr->set_offset(offset);
                offset+=lenth;
                if (tmp_ptr->payload().valid() == "0")  {
                    DLOG("加载到无效消息：%s", tmp_ptr->payload().body().c_str());
                    continue;
                }
                result.push_back(tmp_ptr);
            }
            return true;
        }
        //向队列数据文件中插入消息
        bool insert(const std::string &filename, MessagePtr &msg){
            //插入一个消息的数据，就是插入其有效载荷，需要将有效载荷的长度先插入
            FileHelper _date_file_helper(filename);
            std::string date_body=msg->payload().SerializeAsString();//将有效载荷变成二进制字符串
            //获取有效载荷长度,将其插入进有效位置 
            size_t lenth=date_body.size();
            size_t offset=_date_file_helper.size();//获取当前文件大小，当成偏移量
            //先插入有效长度
            bool ret=_date_file_helper.write((char*)&lenth,offset,sizeof(size_t));
            if(ret == false){
                DLOG("向队列数据文件写入数据长度失败");
                return false;
            }
            //更新msg中的信息
            msg->set_offset(offset+sizeof(size_t));
            msg->set_length(lenth);
            //更新offset
            offset+=sizeof(size_t);
            //插入有效载荷
            ret = _date_file_helper.write(date_body.c_str(),offset,lenth);
            if(ret == false){
                DLOG("向队列数据文件写入数据失败");
                return false;
            }
            return true;
        }
        
        private:
        std::string _queue_name;
        std::string _date_file;
        std::string _tmp_date_file;
    };
class QueueMessage{
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string &basedir, const std::string &qname) : _mapper(basedir, qname),
                                                                       _qname(qname), _valid_count(0), _total_count(0)
        {

        }
        bool recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc();
            for (auto &x : _msgs)
            {
                _durable_msgs.insert(std::make_pair(_qname, x));
            }
            _valid_count = _total_count = _msgs.size();
            return true;
        }

        bool insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            // 1. 构造消息对象
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                // 如果bp存在，则使用bp的原始信息进行对消息的初始化，但消息的可持久化并不是只看消息本身，还要看队列是否持久化
                DeliverMode mode = queue_is_durable ? bp->deliver_mode() : DeliverMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_deliver_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                // 如果bp不存在，则需要我们自己初始化消息的相关信息
                DeliverMode mode = queue_is_durable  ? DeliverMode::DURABLE : DeliverMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_deliver_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            std::unique_lock<std::mutex> lock(_mutex);
            if (msg->payload().properties().deliver_mode() == DeliverMode::DURABLE)
            {
                msg->mutable_payload()->set_valid("1"); // 在持久化存储中表示是有效消息
                bool ret = _mapper.insert(msg);
                if (ret == false)
                {
                    DLOG("持久化存储失败");
                    return false;
                }
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                _valid_count++;
                _total_count++;
            }
            _msgs.push_back(msg);
            return true;
        }
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_msgs.size() == 0)
            {
                return MessagePtr();
            }
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto x = _waitack_msgs.find(msg_id);
            if (x == _waitack_msgs.end())
            {
                DLOG("要确认的消息[%s]并不存在", msg_id);
                return true;
            }
            if (x->second->payload().properties().deliver_mode() == DeliverMode::DURABLE)
            {
                bool ret = _mapper.remove(x->second);
                if (ret == false)
                {
                    DLOG("持久化消息[%s]删除失败", msg_id.c_str());
                }
                _valid_count--;
                _durable_msgs.erase(msg_id);
                // 一旦成功删除，则进行垃圾回收
                if(GCCheck()){
                    gc();
                }
                
            }
            _waitack_msgs.erase(x);
            return true;
        }
        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }
        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }

    private:
        bool GCCheck()
        {
            // 持久化的消息总量大于2000， 且其中有效比例低于50%则需要持久化
            if (_total_count > 2000 && _valid_count * 10 / _total_count < 5)
            {
                return true;
            }
            return false;
        }
        void gc()
        {

            std::list<MessagePtr> msgs = _mapper.gc();
            for (auto &x : msgs)
            {
                auto it = _durable_msgs.find(x->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    _msgs.push_back(x);
                    _durable_msgs.insert(std::make_pair(x->payload().properties().id(), x));
                }
                it->second->set_offset(x->offset());
                it->second->set_length(x->length());
            }
            _valid_count = _total_count = msgs.size();
        }

    private:
        std::mutex _mutex;
        std::string _qname;
        size_t _valid_count;
        size_t _total_count;
        MessageMapper _mapper;
        std::list<MessagePtr> _msgs;                               // 待推送消息,保存在内存当中
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息hash
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息hash
    };
class MessageManager{
    public:
    using ptr = std::shared_ptr<MessageManager>;
    MessageManager(const std::string& basedir):_basedir(basedir){}
    void clear()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        for (auto &qmsg : _queue_msgs){
            qmsg.second->clear();
        }
    }
    void initQueueMessage(const std::string &qname){
        QueueMessage::ptr new_qm;
        //先判断队列消息map中是否有这个队列
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it != _queue_msgs.end())
            {
                DLOG("该消息队列[%s]已经存在", qname.c_str());
                return;
            }
            new_qm= std::make_shared<QueueMessage>(_basedir, qname);
            _queue_msgs.insert(std::make_pair(qname,new_qm));
        }
        new_qm->recovery();
        return ;
    }
    void destroyQueueMessage(const std::string &qname){
        QueueMessage::ptr new_qm;;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("该消息队列[%s]不存在", qname);
                return;
            }
            new_qm=it->second;
            _queue_msgs.erase(qname);
        }
        new_qm->clear();
    }
    bool insertMessage(const std::string &qname, BasicProperties *bp, const std::string &body, DeliverMode flag){
        QueueMessage::ptr new_qm;;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("该消息队列[%s]不存在", qname);
                return false;
            }
            new_qm=it->second;
        }
        bool ret = true;
        if(flag == UNDURABLE) ret = false;
        new_qm->insert(bp,body,ret);
        return true;
    } 
    MessagePtr front(const std::string &qname){
        QueueMessage::ptr new_qm;;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("该消息队列[%s]不存在", qname);
                return MessagePtr();
            }
            new_qm=it->second;
        }
        return new_qm->front();
    }
    void ack(const std::string &qname, const std::string &msg_id) {
        QueueMessage::ptr new_qm;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end()) {
                DLOG("确认队列%s消息%s失败,没有找到消息管理句柄!", qname.c_str(), msg_id.c_str());
                return ;
            }
            new_qm = it->second;
        }
        new_qm->remove(msg_id);
        return ;
    }
    size_t getable_count(const std::string &qname) {
        QueueMessage::ptr new_qm;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end()) {
                DLOG("获取队列%s待推送消息数量失败：没有找到消息管理句柄!", qname.c_str());
                return 0;
            }
            new_qm = it->second;
        }
        return new_qm->getable_count();
    }
    size_t total_count(const std::string &qname) {
        QueueMessage::ptr new_qm;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end()) {
                DLOG("获取队列%s总持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                return 0;
            }
            new_qm = it->second;
        }
        return new_qm->total_count();
    }
    size_t durable_count(const std::string &qname) {
        QueueMessage::ptr new_qm;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end()) {
                DLOG("获取队列%s有效持久化消息数量失败：没有找到消息管理句柄!", qname.c_str());
                return 0;
            }
            new_qm = it->second;
        }
        return new_qm->durable_count();
    }
    size_t waitack_count(const std::string &qname) {
        QueueMessage::ptr new_qm;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end()) {
                DLOG("获取队列%s待确认消息数量失败：没有找到消息管理句柄!", qname.c_str());
                return 0;
            }
            new_qm = it->second;
        }
        return new_qm->waitack_count();
    }
    private:
    std::mutex _mutex;
    std::string _basedir;
    std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;
    };
}
#endif